package yao.mq.p2p;



import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消息的消费者（接受者）
 * 
 * @author liang
 *
 */
public class JMSConsumer {

	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;// 默认连接用户名
	private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;// 默认连接密码
	private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;// 默认连接地址
	private static final Object t1_lock = new Object();	
	private static final Object t2_lock = new Object();	
	public static void main(String[] args) {
		ConnectionFactory connectionFactory;// 连接工厂
		Connection connection = null;// 连接

		Session session;// 会话 接受或者发送消息的线程
		Destination destination;// 消息的目的地

		final MessageConsumer messageConsumer;// 消息的消费者

		// 实例化连接工厂
		connectionFactory = new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD,
				JMSConsumer.BROKEURL);

		try {
			// 通过连接工厂获取连接
			connection = connectionFactory.createConnection();
			// 启动连接
			connection.start();
			// 创建session
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			// 创建一个连接HelloWorld的消息队列
			destination = session.createQueue("HelloWorld1");
			// 创建消息消费者1
			messageConsumer = session.createConsumer(destination);
			
			final Thread t1 = new Thread(new Runnable() {							
				@Override
				public void run() {
					while (true) {
						try {
							synchronized (t1_lock) {
								TextMessage textMessage = (TextMessage) messageConsumer.receive(100000);
								if (textMessage != null) {
									System.out.println("消费者1：收到的消息:" + textMessage.getText());
								}
								Thread.currentThread().sleep(3000);
							}							
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
				}
			});
			t1.start();
			
			// 创建消息消费者2
			final MessageConsumer	messageConsumer2 = session.createConsumer(destination);
			final Thread t2 = new Thread(new Runnable() {				
				@Override
				public void run() {
					while (true) {
						try {
							synchronized (t2_lock) {
								TextMessage textMessage = (TextMessage) messageConsumer2.receive(100000);
								if (textMessage != null) {
									System.out.println("消费者2：收到的消息:" + textMessage.getText());
								}
								Thread.currentThread().sleep(2000);
							}							
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
					
				}
			});
			t2.start();
			
			

		} catch (JMSException e) {
			e.printStackTrace();
		}

	}
}